package com.cn.daimajiangxin.flink.sink;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.connector.elasticsearch.sink.Elasticsearch7SinkBuilder;
import org.apache.flink.connector.elasticsearch.sink.ElasticsearchSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;

import java.util.Map;

public class ElasticsearchSinkDemo {
    private static final ObjectMapper objectMapper = new ObjectMapper();

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);

        DataStream<String> stream = env.fromData(
                "{\"id\":\"1\",\"name\":\"Flink\",\"category\":\"framework\"}",
                "{\"id\":\"2\",\"name\":\"Elasticsearch\",\"category\":\"database\"}");

        // 配置Elasticsearch节点
        HttpHost httpHost = new HttpHost("192.168.0.199", 9200, "http");

        // 创建Elasticsearch Sink
        ElasticsearchSink<String> sink = new Elasticsearch7SinkBuilder<String>()
                .setBulkFlushMaxActions(10) // 批量操作数量
                .setBulkFlushInterval(5000) // 批量刷新间隔（毫秒）
                .setHosts(httpHost)
                .setConnectionRequestTimeout(60000) // 连接请求超时时间
                .setConnectionTimeout(60000) // 连接超时时间
                .setSocketTimeout(60000) // Socket 超时时间
                .setEmitter((element, context, indexer) -> {
                    try {
                        Map<String, Object> json = objectMapper.readValue(element, Map.class);
                        IndexRequest request = Requests.indexRequest()
                                .index("flink_documents")
                                .id((String) json.get("id"))
                                .source(json);
                        indexer.add(request);
                    } catch (Exception e) {
                        // 处理解析异常
                        System.err.println("Failed to parse JSON: " + element);
                    }
                })
                .build();

        // 添加Sink
        stream.sinkTo(sink);

        env.execute("Elasticsearch Sink Demo");
    }
}